package cn.piesat.client.listener;

import cn.piesat.client.constant.cache.LocalCache;
import cn.piesat.client.service.ClientContext;
import cn.piesat.client.utils.ThreadPoolLock;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * kafka任务消息接受
 */
@Component
public class KafkaTask01Listener {

    private static Logger logger = LoggerFactory.getLogger(KafkaTask01Listener.class);

    private ThreadPoolLock threadPoolLock;

    @Autowired
    private ClientContext clientContext;

    private String taskThreadPoolSize;

    @Value("${dms.task.thread.pool.size01}")
    public void setTaskThreadPoolSize(String taskThreadPoolSize) {
        this.taskThreadPoolSize = taskThreadPoolSize;
    }

    @PostConstruct
    public void initLock(){
        threadPoolLock = new ThreadPoolLock(taskThreadPoolSize);
        LocalCache.kafkaListenerPoolLock01 = threadPoolLock;
    }

    @KafkaListener(topics = "${spring.kafka.consumer.exec_01.topic}", groupId = "${spring.kafka.consumer.exec_01.group}")
    public void taskHandle1(ConsumerRecord<String, String> record, Consumer<String, String> consumer) {
        logger.info("接收到任务执行消息offset:{},value:{}", record.offset(), record.value());
        try {
            threadPoolLock.record();
            clientContext.doProcess01(record.value());
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            consumer.commitSync();
            threadPoolLock.doWait();
        }
    }


}
